package com.inspinia.base.mq.rabbitmq.service;

import com.inspinia.base.mq.rabbitmq.config.RabbitConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Resource;

/**
 * 消息接收监听
 * @author Veblen
 */
@Slf4j
public class RabbitMqMessageReceiver {

    private String queueName;

    @Resource
    RabbitMqConsumerRegister register;

    /**
     * 消息处理机，只有handleBiz方法
     */
    private RabbitMqHandler handler;

    public RabbitMqMessageReceiver(RabbitMqConsumerRegister register, String taskQueueName, RabbitMqHandler handler) {
        this.register = register;
        this.queueName = taskQueueName;
        this.handler = handler;
    }

    /**
     * 正常接收这个队列的消息
     */
    public void normalReceive() {
        normalReceive(RabbitConfig.QOS);
    }

    /**
     * 正常接收这个队列的消息
     *
     * @param qos
     * 除对时效要求紧急且为小任务，其他情况请勿将qos设置为0
     */
    public void normalReceive(int qos) {
        register.addNormalCounsumer(queueName, new RabbitMessageHandler(), false, qos);
    }

    /**
     * 正常接收这个队列的消息
     */
    public void normalReceiveAutoAck() {
        normalReceiveAutoAck(RabbitConfig.QOS);
    }

    /**
     * 正常接收这个队列的消息
     *
     * @param qos
     *            除对时效要求紧急且为小任务，其他情况请勿将qos设置为0
     */
    public void normalReceiveAutoAck(int qos) {
        register.addNormalCounsumer(queueName, new RabbitMessageHandler(), true, qos);
    }

    /**
     * 延迟接收这个队列的消息
     * <p>
     * 如果autoAck为true，说明这个消息会自动确认（handle方法的true和false无效），如果需要重试，则需要在代码中执行消息的重试逻辑（重新发送消息等）
     * <p>
     * 如果autoAck为false，说明这个消息由业务代码确认，就是handle方法的true或false确认
     * <p>
     * 除对时效要求紧急且为小任务，其他情况请勿将qos设置为0
     */
    public void delayReceive(boolean autoAck, int qos) {
        register.addDelayConsumer(queueName, "", new RabbitMessageHandler(), autoAck, qos);
    }

    public class RabbitMessageHandler implements RabbitMqConsumerRegister.MessageHandler {
        @Override
        public void handle(final Channel channel, final String message, long deliveryTag, boolean autoAck) throws InterruptedException {
            register.submitTask(() -> {
                log.info("Received[{}]:{}", deliveryTag, message);
                try {
                    // 处理消息并获取消息处理结果，true表示处理成功，false表示处理失败
                    boolean isMessageHandled = handler.handleBiz(message);
                    // 如果是自动应答的，不需要进行手动ack，这样会导致消息丢失
                    if (!autoAck) {
                        // 消息处理成功，进行处理成功后的流程
                        if (isMessageHandled) {
                            channel.basicAck(deliveryTag, false);
                            log.info("Handle finished[{}]", deliveryTag);
                            // 判断是否需要重试
                        } else {
                            // 消息处理失败，进行处理失败后的流程
                            register.requeue(channel, deliveryTag);
                            log.info("消息消费失败，等待自动超时后处理");
                        }
                    }
                } catch (Exception e) {
                    log.error("消息处理失败", e);
                    // 若处理消息中产生了异常，这里采用保守方式，走消息处理失败流程
                    if (!autoAck) {
                        //只有没有开启ack的,才能重新入队
                        register.requeue(channel, deliveryTag);
                    }
                }
            });
        }
    }
}